package org.javaboy.redis_stream_demo.service;

import org.javaboy.redis_stream_demo.config.RedisQueueConfig;
import org.javaboy.redis_stream_demo.model.User;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.stream.StreamListener;

/**
 * 消息监听器
 */
public class AsyncStreamListener implements StreamListener<String, ObjectRecord<String, User>> {

    private String groupName;
    private String consumerName;

    RedisTemplate redisTemplate;

    public AsyncStreamListener(String groupName, String consumerName, RedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
        this.groupName = groupName;
        this.consumerName = consumerName;
    }

    @Override
    public void onMessage(ObjectRecord<String, User> message) {
        RecordId id = message.getId();
        User value = message.getValue();
        System.out.println(groupName + ":" + consumerName + "消费了id为" + id.getValue() + "的数据,数据内容是：" + value);
        //手动 ACK
//        redisTemplate.opsForStream().acknowledge(RedisQueueConfig.QUEUE_NAME, groupName, id);
    }
}
